package com.cloudeye.business.job.leadership;

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by lafangyuan on 2021/6/21.
 * 原理：使用了Curator内置的InterProcessMutex分布式锁来实现
 */
public class LeaderShipSelectorAware {

    public static final String ZOOKEEPERSTRING = "localhost:2181";
    static String LOCK_PATH = "/zk/leader";


    static class MyLeaderClient extends LeaderSelectorListenerAdapter implements Closeable{

        private final String name;

        private final LeaderSelector leaderSelector;

        public MyLeaderClient(CuratorFramework client,String path ,String name) {
            this.name = name;
            this.leaderSelector = new LeaderSelector(client,path,this);
            this.leaderSelector.autoRequeue();
        }

        public void start(){
            this.leaderSelector.start();
        }
        @Override
        public void close() throws IOException {
            this.leaderSelector.close();
        }

        @Override
        public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
            // 抢到leader权限后sleep一段时间，并释放leader权限
            final int waitSeconds = (int) (5 * Math.random()) + 1;

            System.out.println(name + " is now the leader. Waiting " + waitSeconds + " seconds...");
            System.out.println("我现在是Leader:"+name);
            try {
                Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
            } catch (InterruptedException e) {
                System.err.println(name + " was interrupted.");
                Thread.currentThread().interrupt();
            } finally {
                System.out.println(name + " relinquishing leadership.\n");
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        List<CuratorFramework> clientsList = Lists.newArrayListWithCapacity(10);
        //启动10个zk客户端，每几秒进行一次leader选举
        for (int i = 0; i < 10; i++) {
            CuratorFramework client = getZkClient();
            clientsList.add(client);
            MyLeaderClient exampleClient = new MyLeaderClient(client, LOCK_PATH, "client_" + i);
            exampleClient.start();
        }
        //sleep 以观察抢主过程
        Thread.sleep(Integer.MAX_VALUE);
    }
    private static CuratorFramework getZkClient() {
        String zkServerAddress = "127.0.0.1:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        return zkClient;
    }
}
